package statebackend;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class statebackend {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000); // 设置检查点
        env.setStateBackend(new FsStateBackend("hdfs://192.168.43.140:9000/flink/wc5"));
    //    使用工厂模式
        env.socketTextStream("192.168.43.51",1234).flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(" ");
                System.out.println(words);
                for(String w:words){
                    collector.collect(new Tuple2<String, Integer>(w,1));
                }
            }
        }).keyBy(0).sum(1).print();

        env.execute("fsstatebackend");
    }
}
